package util;

import com.google.common.base.CaseFormat;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.phoenix.schema.TypeMismatchException;

import java.io.Serializable;
import java.lang.reflect.Field;
import java.sql.*;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class JoinHbaseUtil implements Serializable {

    private static Connection connection;
    private static ThreadPoolExecutor threadPoolExecutor;
    public static final String PHOENIX_DRIVER = "org.apache.phoenix.jdbc.PhoenixDriver";

    public static final String PHOENIX_SERVER = "jdbc:phoenix:node101:2181";


    public static <LEFT, RIGHT, OUT> SingleOutputStreamOperator<OUT> unorderedWait(
            DataStream<LEFT> in,
            MyDIMRequestIO<LEFT, RIGHT, OUT> func) {


        return AsyncDataStream.unorderedWait(in, new RichAsyncFunction<LEFT, OUT>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                Properties properties = new Properties();
                properties.setProperty("phoenix.schema.isNamespaceMappingEnabled", "true");
                properties.setProperty("phoenix.schema.mapSystemTablesToNamespace", "true");
                Class.forName(PHOENIX_DRIVER);
                connection = DriverManager.getConnection(PHOENIX_SERVER, properties);
                threadPoolExecutor = MyThreadPoolUtil.getThreadPool();
            }

            @Override
            public void asyncInvoke(LEFT input, ResultFuture<OUT> resultFuture) throws Exception {
                threadPoolExecutor.submit(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            Object key = func.selectLeftKey(input);
                            if (key == null || "".equals(key)) {
                                key = "1";
                            }
                            String replace = func.buildSql().replace("?", key.toString());

                            ArrayList<RIGHT> resultList = new ArrayList<>();
                            PreparedStatement preparedStatement = connection.prepareStatement(replace);
                            ResultSet resultSet = preparedStatement.executeQuery();
                            ResultSetMetaData metaData = resultSet.getMetaData();
                            int columnCount = metaData.getColumnCount();
                            while (resultSet.next()) {
                                RIGHT t = func.selectRightClass().newInstance();
                                for (int i = 1; i < columnCount + 1; i++) {
                                    Object value = resultSet.getObject(i);
                                    // 数据库里的字段名
                                    String columnName = metaData.getColumnName(i);
                                    if (columnName.contains("_vv2") || columnName.contains("_VV2")) {
                                        columnName = columnName.replace("_vv2", "");
                                        columnName = columnName.replace("_VV2", "");
                                    }
                                    if (true) {
                                        columnName = CaseFormat.LOWER_UNDERSCORE
                                                .to(CaseFormat.LOWER_CAMEL, columnName.toLowerCase());
                                    }
                                    Field[] declaredFields = func.selectRightClass().getDeclaredFields();

                                    for (Field field : declaredFields) {
                                        if (field.getName().equalsIgnoreCase(columnName)) {
                                            BeanUtils.setProperty(t, columnName, value);
                                            break;
                                        }
                                    }
                                }
                                resultList.add(t);
                            }
                            preparedStatement.close();
                            resultSet.close();

                            if (resultList == null || resultList.size() == 0) {
                                System.out.println("ERROR : sql key is null : " + replace);
                                resultList.add(func.selectRightClass().newInstance());
                            }
                            OUT join = func.join(input, resultList.get(0));

                            resultFuture.complete(Collections.singleton(join));

                        } catch (TypeMismatchException e) {
                            try {
                                Object key = func.selectLeftKey(input);
                                if (key == null || "".equals(key)) {
                                    key = "1";
                                }
                                String replace = func.buildSql().replace("?", "'" + key.toString() + "'");


                                ArrayList<RIGHT> resultList = new ArrayList<>();
                                PreparedStatement preparedStatement = connection.prepareStatement(replace);
                                ResultSet resultSet = preparedStatement.executeQuery();
                                ResultSetMetaData metaData = resultSet.getMetaData();
                                int columnCount = metaData.getColumnCount();
                                while (resultSet.next()) {
                                    RIGHT t = func.selectRightClass().newInstance();
                                    for (int i = 1; i < columnCount + 1; i++) {
                                        Object value = resultSet.getObject(i);
                                        // 数据库里的字段名
                                        String columnName = metaData.getColumnName(i);
                                        if (true) {
                                            columnName = CaseFormat.LOWER_UNDERSCORE
                                                    .to(CaseFormat.LOWER_CAMEL, columnName.toLowerCase());
                                        }
                                        Field[] declaredFields = func.selectRightClass().getDeclaredFields();

                                        for (Field field : declaredFields) {
                                            if (field.getName().equalsIgnoreCase(columnName)) {
                                                BeanUtils.setProperty(t, columnName, value);
                                                break;
                                            }
                                        }
                                    }
                                    resultList.add(t);
                                }
                                preparedStatement.close();
                                resultSet.close();
                                if (resultList == null || resultList.size() == 0) {
                                    resultList.add(func.selectRightClass().newInstance());
                                    System.out.println("ERROR : sql key is null : " + replace);
                                }
                                OUT join = func.join(input, resultList.get(0));

                                resultFuture.complete(Collections.singleton(join));
                            } catch (Exception e2) {
                                System.out.println("ERROR :  " + e2.getMessage());
//                        e2.printStackTrace();
                            }
                        } catch (Exception e) {
                            System.out.println("ERROR :  " + e.getMessage());
//                    e.printStackTrace();
                        }
                    }
                });

            }
        }, 50, TimeUnit.SECONDS).returns(func.selectOutClass());


    }

}

class MyThreadPoolUtil {

    private static ThreadPoolExecutor threadPoolExecutor = null;

    private MyThreadPoolUtil() {
    }

    public static ThreadPoolExecutor getThreadPool() {

        if (threadPoolExecutor == null) {
            synchronized (MyThreadPoolUtil.class) {
                if (threadPoolExecutor == null) {
                    threadPoolExecutor = new ThreadPoolExecutor(8,
                            16,
                            1L,
                            TimeUnit.MINUTES,
                            new LinkedBlockingDeque<>());
                }
            }
        }

        return threadPoolExecutor;
    }
}

